package org.sean.framework.kafka.logging.appender;

import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.Appender;
import ch.qos.logback.core.spi.AppenderAttachableImpl;
import org.apache.kafka.clients.KafkaClient;
import org.apache.kafka.clients.producer.internals.Sender;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * Kafka Appender
 *
 * @param <E> E
 * @author xielei
 */
public class KafkaAppender<E> extends AbstractKafkaAppender<E> {

    private static final String KAFKA_LOGGER_PREFIX = KafkaClient.class.getPackage().getName();
    private static final String KAFKA_LOGGER_SENDER = Sender.class.getName();
    private final AppenderAttachableImpl<E> aai = new AppenderAttachableImpl<>();
    private final ConcurrentLinkedQueue<E> queue = new ConcurrentLinkedQueue<>();
    private LazyProducer lazyProducer = null;

    @Override
    public void doAppend(E e) {
        if (e instanceof ILoggingEvent) {
            String loggerName = ((ILoggingEvent) e).getLoggerName();
            // 防止循环打印
            if (loggerName.equals(KAFKA_LOGGER_SENDER)) {
                return;
            }
            if (loggerName.startsWith(KAFKA_LOGGER_PREFIX)) {
                deferAppend(e);
            } else {
                ensureDeferredAppends();
                super.doAppend(e);
            }
        } else {
            super.doAppend(e);
        }
    }

    @Override
    public void start() {
        lazyProducer = new LazyProducer();
        super.start();
    }

    @Override
    public void addAppender(Appender<E> newAppender) {
        aai.addAppender(newAppender);
    }

    @Override
    public Iterator<Appender<E>> iteratorForAppenders() {
        return aai.iteratorForAppenders();
    }

    @Override
    public Appender<E> getAppender(String name) {
        return aai.getAppender(name);
    }

    @Override
    public boolean isAttached(Appender<E> appender) {
        return aai.isAttached(appender);
    }

    @Override
    public void detachAndStopAllAppenders() {
        aai.detachAndStopAllAppenders();
    }

    @Override
    public boolean detachAppender(Appender<E> appender) {
        return aai.detachAppender(appender);
    }

    @Override
    public boolean detachAppender(String name) {
        return aai.detachAppender(name);
    }

    @Override
    protected void append(E e) {
        final String payload = encoder.doEncode(e);
        final String key = keyingStrategy.createKey(e);
        KafkaTemplate<Object, Object> producer = lazyProducer.get();
        if (producer == null) {
            return;
        }
        deliveryStrategy.send(producer, topic, key, payload);
    }

    /**
     * 消息插入队列
     */
    private void deferAppend(E event) {
        queue.add(event);
    }

    /**
     * 处理队列
     */
    private void ensureDeferredAppends() {
        E event;

        while ((event = queue.poll()) != null) {
            super.doAppend(event);
        }
    }

    /**
     * Lazy initializer for producer, patterned after commons-lang.
     *
     * @see <a href="https://commons.apache.org/proper/commons-lang/javadocs/api-3.4/org/apache/commons/lang3/concurrent/LazyInitializer.html">LazyInitializer</a>
     */
    private class LazyProducer {

        private volatile KafkaTemplate<Object, Object> producer;

        public KafkaTemplate<Object, Object> get() {
            KafkaTemplate<Object, Object> result = producer;
            if (result == null) {
                synchronized (this) {
                    result = producer;
                    if (result == null) {
                        producer = result = kafkaTemplate();
                    }
                }
            }

            return result;
        }

    }

}
